package sos.entrega2013.mr1.noviembre.bo.fraude;

import java.io.IOException;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.StringTokenizer;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.MapReduceBase;
import org.apache.hadoop.mapred.Mapper;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reporter;

import sos.entrega2013.mr1.noviembre.beans.CDR;

public class FraudeMap extends MapReduceBase 
	implements Mapper<LongWritable, Text, Text, CDR>{

	//
	private SimpleDateFormat formatoDeFecha = new SimpleDateFormat("yyyyMMddHHmmss");
	private Text 			 tLlamante 		= new Text();
	private Text			 tLlamado		= new Text();
	private CDR				 cdrLlamada 	= new CDR();
	private Date		 	 dFecha		 	= null;
	private String 			 sNTelfLlamante	= null;
	private String 			 sNtelfLlamado	= null;
	private int 			 iXllamante;
	private int 			 iXllamado;
	private int 			 iYllamante;
	private int				 iYllamado;
	private int 			 iDuracion;
	
	@Override
	public void map(LongWritable key, Text value,OutputCollector<Text, CDR> output, Reporter reporter)
		throws IOException {
		//
		String sLine 				= value.toString();
		StringTokenizer tokenizer 	= new StringTokenizer(sLine,",");
		/* Leemos los datos del fichero CVS*/
		//telefono del llamante
		this.sNTelfLlamante	=	tokenizer.nextToken().trim();
		//telefono del llamado
		this.sNtelfLlamado 	= tokenizer.nextToken().trim();
		//obtenemos la fecha 
		try {
			this.dFecha=formatoDeFecha.parse(tokenizer.nextToken().trim());
		} catch (ParseException e) {e.printStackTrace();}
		//duracion de la llamada
		this.iDuracion	=	Integer.valueOf(tokenizer.nextToken().trim());
		//coordenadas X,Y del llamante
		this.iXllamante	= 	Integer.valueOf(tokenizer.nextToken().trim());
		this.iYllamante	=	Integer.valueOf(tokenizer.nextToken().trim());
		//coordenadas X,Y del Llamado
		this.iXllamado	=	Integer.valueOf(tokenizer.nextToken().trim());
		this.iYllamado	=	Integer.valueOf(tokenizer.nextToken().trim());
		//--------------------------------------------------------------
		//Asignamos los datos al Text
		this.tLlamante.set(this.sNTelfLlamante);
		this.tLlamado.set(this.sNtelfLlamado);
		//Asignamos los datos al bean
		this.cdrLlamada.setLlamante(this.sNTelfLlamante);
		this.cdrLlamada.setLlamado(this.sNtelfLlamado);
		this.cdrLlamada.setFecha(this.dFecha);
		this.cdrLlamada.setDuracion(this.iDuracion);
		this.cdrLlamada.setCoordXllamante(iXllamante);
		this.cdrLlamada.setCoordYllamante(this.iYllamante);
		this.cdrLlamada.setCoordXllamado(this.iXllamado);
		this.cdrLlamada.setCoordYllamado(this.iYllamado);
		//asignamos la salida
		output.collect(this.tLlamante, this.cdrLlamada);
		output.collect(this.tLlamado, this.cdrLlamada);
		
	}

}
